package com.sunye.elasticsearch.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.ArrayUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.sunye.elasticsearch.entity.*;
import com.sunye.elasticsearch.mapper.EsDogMapper;
import com.sunye.elasticsearch.repository.EsDogIndexRepository;
import com.sunye.elasticsearch.service.EsDogSellingPointService;
import com.sunye.elasticsearch.service.EsDogService;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.lucene.search.function.FieldValueFactorFunction;
import org.elasticsearch.common.lucene.search.function.FunctionScoreQuery;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.functionscore.*;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.aggregations.*;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.nested.ParsedNested;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.CardinalityAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedMax;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.sort.GeoDistanceSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author sunye
 * @since 2022-07-13
 */
@Slf4j
@Service
public class EsDogServiceImpl extends ServiceImpl<EsDogMapper, EsDog> implements EsDogService {
    protected static final RequestOptions COMMON_OPTIONS;
    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();

        //默认缓冲限制为100MB，此处修改为30MB。
        builder.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    @Autowired
    private EsDogIndexRepository esDogIndexRepository;

    @Autowired
    @Qualifier("restHighLevelClient")
    protected RestHighLevelClient client;


    /**
     * ES
     */
    @Autowired
    private ElasticsearchOperations restTemplate;

    @Autowired
    private EsDogSellingPointService esDogSellingPointService;

    @Override
    public Iterable selectAll() {
        Iterable<EsDogIndex> list = esDogIndexRepository.findAll();
        return list;
    }

    @Override
    public void changeSellingpoint(ChangeSelling changeSelling) {
        String id = changeSelling.getId();
        String advantageDescribe = changeSelling.getAdvantageDescribe();
        EsDogSellingPoint esDogSellingPoint = esDogSellingPointService.getEsDogSellingPointById(id);
        esDogSellingPoint.setAdvantageDescribe(advantageDescribe);
        if(esDogSellingPoint !=null){
            UpdateByQueryRequest request = new UpdateByQueryRequest("dog_index");
            request.setQuery(QueryBuilders.nestedQuery("sellingPointList", QueryBuilders.termQuery("sellingPointList.sellingPointId", id), ScoreMode.None));
            //request.setScript(new Script("for(e in ctx._source.sellingPointList){if (e.sellingPointId== " + id + ") {e.advantageDescribe = '" + advantageDescribe + "';}}"));
            request.setScript(new Script(ScriptType.INLINE,"painless","for(e in ctx._source.sellingPointList){if (e.sellingPointId == " + id + ") {e.advantageDescribe = '" + advantageDescribe + "';}}", Collections.emptyMap()));
            log.info("dsl{}",request.toString());
            try {
                BulkByScrollResponse bulkByScrollResponse = client.updateByQuery(request, RequestOptions.DEFAULT);
                System.out.println("bulkByScrollResponse = " + bulkByScrollResponse);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public NativeSearchQueryBuilder searchDogsByEs(EsDogSearchDTO dogSearchDTO, PageVO<EsDogIndex> page) {
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        if (page != null) {
            int pageNumber = page.getPageNum() - 1;
            if (pageNumber < 0) {
                pageNumber = 0;
            }
            Pageable pageable = PageRequest.of(pageNumber, page.getPageSize());
            //分页 .withPageable
            nativeSearchQueryBuilder.withPageable(pageable);
        }

        //查询参数非空判定
        if (dogSearchDTO != null) {
            //boolQuery 先构造boolQuery对象  添加逻辑判断条件
            //过滤条件
            BoolQueryBuilder filterBuilder = QueryBuilders.boolQuery();
            //对查询条件进行处理
            this.commonSearch(filterBuilder, dogSearchDTO);
            //条件 .withQuery
            nativeSearchQueryBuilder.withQuery(filterBuilder);
            //排序 .withSort
            if (StringUtils.isNotBlank(dogSearchDTO.getPosition())) {
                String[] split = dogSearchDTO.getPosition().split("_");
                Double lat = Double.parseDouble(split[0]);
                Double lon = Double.parseDouble(split[1]);
                //按照距离近的排序
                nativeSearchQueryBuilder.withSort(SortBuilders.geoDistanceSort("location",lat,lon).unit(DistanceUnit.KILOMETERS).order(SortOrder.ASC));
            }
            if(StringUtil.isNotBlank(dogSearchDTO.getSort()) ){
                //排序字段排序（价格和年龄）
                String[] fileds = dogSearchDTO.getSort().split(",");
                for (String filed : fileds) {
                    String[] sorts = filed.split("_");
                    nativeSearchQueryBuilder.withSort(SortBuilders.fieldSort(sorts[0]).order("asc".equals(sorts[1]) ? SortOrder.ASC : SortOrder.DESC));
                }
            }else{
                //默认按照销量排序
                nativeSearchQueryBuilder.withSort(SortBuilders.fieldSort("salesNumber").order(SortOrder.DESC));
            }
        }
        return nativeSearchQueryBuilder;
    }



    /**
     * 查询属性处理
     *
     * @param filterBuilder 过滤构造器
     * @param searchDTO     查询参数
     */
    private void commonSearch(BoolQueryBuilder filterBuilder, EsDogSearchDTO searchDTO) {
        /**
         * 1.must和filter是一样的。区别是场景不一样。如果结果需要算分就使用must，否则可以考虑使用filter，使查询更高效。
         * 2.termQuery（精确匹配查询:不会进行分词【相当于sql：field = 1 】）
         * 3.wildcardQuery （通配符查询【相当于sql：field like '张_三*'】）
         * 5.Nested Query(嵌套查询)
         */
        /**
         * must 返回的文档必须满足must子句的条件,并且参与计算分值
         *
         * filter 返回的文档必须满足filter子句的条件,但是不会像must一样,参与计算分值
         *
         * should 返回的文档可能满足should子句的条件.在一个bool查询中,如果没有must或者filter,有一个或者多个should子句,那么只要满足一个就可以返回.
         *
         * must_not 返回的文档必须不满足定义的条件
         *
         * 组合查询BoolQueryBuilder
         * must(QueryBuilders) :AND
         * mustNot(QueryBuilders):NOT
         * should: OR
         *
         * matchAllQuery匹配所有
         *
         * termQuery精准匹配，大小写敏感且不支持
         *
         * matchPhraseQuery对中文精确匹配
         *
         * matchQuery("key", Obj) 单个匹配, field不支持通配符, 前缀具高级特性
         *
         * multiMatchQuery("text", "field1", "field2"..); 匹配多个字段, field有通配符忒行
         */
        //关键字查询
        String keyword = searchDTO.getKeyword();
        if(StringUtils.isNotBlank(keyword)){
            //权重方式查询（不常用）
            //FunctionScoreQueryBuilder filterFunctionBuilders = this.buildKeywordSearch(keyword);
            //filterBuilder.must(filterFunctionBuilders);
            //正常查询（常用  shoule 表示 or）
            BoolQueryBuilder boolQueryBuilder1 = QueryBuilders.boolQuery();
            boolQueryBuilder1.should(QueryBuilders.matchQuery("dogName",keyword));
            boolQueryBuilder1.should(QueryBuilders.matchQuery("introduct",keyword));
            boolQueryBuilder1.should(QueryBuilders.nestedQuery("sellingPointList",QueryBuilders.matchQuery("sellingPointList.advantage",keyword),ScoreMode.None));
            filterBuilder.must(boolQueryBuilder1);
        }else{
            filterBuilder.must(QueryBuilders.matchAllQuery());
        }
        //颜色
        if (StringUtil.isNotBlank(searchDTO.getColor())) {
            filterBuilder.filter(QueryBuilders.wildcardQuery("color","*" + searchDTO.getColor() + "*"));
        }
        //分类(匹配多个字段)
        if(searchDTO.getCategoryId() !=null){
            filterBuilder.filter(QueryBuilders.multiMatchQuery(searchDTO.getCategoryId(),"categoryFirstId","categorySecondId"));
        }
        //是否为推荐狗
        if (searchDTO.getRecommend() != null) {
            filterBuilder.filter(QueryBuilders.termQuery("recommend", searchDTO.getRecommend()));
        }
        //卖点-->嵌套查询 (and的查询)
        if (searchDTO.getSellingPointIds() != null && !searchDTO.getSellingPointIds().isEmpty()) {
            for (String pointId : searchDTO.getSellingPointIds()) {
                filterBuilder.filter(QueryBuilders.nestedQuery("sellingPointList", QueryBuilders.termsQuery("sellingPointList.sellingPointId",pointId), ScoreMode.None));
            }
        }
        //sku-->嵌套查询(or的查询)
        if (searchDTO.getSkuIds() != null && !searchDTO.getSkuIds().isEmpty()) {
            BoolQueryBuilder boolQueryBuilder2 = QueryBuilders.boolQuery();
            for (String skuId : searchDTO.getSkuIds()) {
                boolQueryBuilder2.should(QueryBuilders.nestedQuery("skuList", QueryBuilders.termsQuery("skuList.id", skuId), ScoreMode.None));
            }
            filterBuilder.filter(boolQueryBuilder2);
        }
        //价格区间判定
        if (StringUtils.isNotEmpty(searchDTO.getPrice())) {
            String[] prices = searchDTO.getPrice().split("_");
            if (prices.length == 0) {
                return;
            }
            double min = Convert.toDouble(prices[0], 0.0);
            double max = Integer.MAX_VALUE;

            if (prices.length == 2) {
                max = Convert.toDouble(prices[1], Double.MAX_VALUE);
            }
            filterBuilder.filter(QueryBuilders.rangeQuery("price").from(min).to(max).includeLower(true).includeUpper(true));
        }
        //未上架的商品不显示
        filterBuilder.filter(QueryBuilders.termQuery("saleStatus", "1"));
    }


    /**
     * 关键字查询（通过Weight的方式来实现权重排序  不常用可能查询还有问题）
     *
     * @param keyword 关键字
     * @return 构造查询的集合
     */
    private FunctionScoreQueryBuilder buildKeywordSearch(String keyword) {
        List<FunctionScoreQueryBuilder.FilterFunctionBuilder> filterFunctionBuilders = new ArrayList<>();
        // operator 为 AND 时 需全部分词匹配。为 OR 时 需配置 minimumShouldMatch（最小分词匹配数）不设置默认为1
        //名字匹配
        MatchQueryBuilder goodsNameQuery = QueryBuilders.matchQuery("dogName", keyword).operator(Operator.OR).minimumShouldMatch("2");
        filterFunctionBuilders.add(new FunctionScoreQueryBuilder.FilterFunctionBuilder(goodsNameQuery,
                ScoreFunctionBuilders.weightFactorFunction(10)));
        //简介匹配
        MatchQueryBuilder introductQuery = QueryBuilders.matchQuery("introduct", keyword).operator(Operator.OR).minimumShouldMatch("2");
        filterFunctionBuilders.add(new FunctionScoreQueryBuilder.FilterFunctionBuilder(introductQuery,
                ScoreFunctionBuilders.weightFactorFunction(10)));

        GaussDecayFunctionBuilder skuNoScore = ScoreFunctionBuilders.gaussDecayFunction("salesNumber", 100, 10).setWeight(7);
        FunctionScoreQueryBuilder.FilterFunctionBuilder skuNoBuilder = new FunctionScoreQueryBuilder.FilterFunctionBuilder(goodsNameQuery, skuNoScore);
        filterFunctionBuilders.add(skuNoBuilder);
        //封装到FunctionScoreQueryBuilder里 查询
        FunctionScoreQueryBuilder.FilterFunctionBuilder[] builders = new FunctionScoreQueryBuilder.FilterFunctionBuilder[filterFunctionBuilders.size()];
        filterFunctionBuilders.toArray(builders);
        FunctionScoreQueryBuilder functionScoreQueryBuilder = QueryBuilders.functionScoreQuery(builders)
                .scoreMode(FunctionScoreQuery.ScoreMode.SUM)
                .setMinScore(2);

        return functionScoreQueryBuilder;
    }


    @Override
    public List searchDogsGroupBy(EsDogSearchDTO dogSearchDTO) {
        NativeSearchQueryBuilder searchQueryBuilder = this.searchDogsByEs(dogSearchDTO, null);
        //分组--参考https://blog.csdn.net/qq_16713463/article/details/98498461?utm_medium=distribute.pc_aggpage_search_result.none-task-blog-2~aggregatepage~first_rank_ecpm_v1~rank_v31_ecpm-28-98498461-null-null.pc_agg_new_rank&utm_term=ElasticsearchRepository%20%E5%88%86%E7%BB%84%E8%81%9A%E5%90%88&spm=1000.2123.3001.4430
        if (StringUtil.isNotBlank(dogSearchDTO.getGroupByAge())) {
            //select age, count(*) as ageCount from es_dog WHERE sale_status = '1' AND deleted = '0' group by age; -- age是int不用写keyword
            searchQueryBuilder.addAggregation(AggregationBuilders.terms("ageCount").field("age"));
        }
        //多个字段分组
        if (StringUtil.isNotBlank(dogSearchDTO.getGroupByAgeAndBreed())) {
            //select age, dog_breed, count(*) as breed_count from es_dog WHERE sale_status = '1' AND deleted = '0' group by age, dog_breed;
            AggregationBuilder categoryNameBuilder = AggregationBuilders.terms("breedCount").field("dogBreed.keyword");
            searchQueryBuilder.addAggregation(AggregationBuilders.terms("ageCount").field("age").subAggregation(categoryNameBuilder));
        }
        //分组取最大值(max写在主查询上)
        if (StringUtil.isNotBlank(dogSearchDTO.getGroupByMaxAge())) {
            //select dog_breed, max(age) as max_age from es_dog WHERE sale_status = '1' AND deleted = '0' group by dog_breed;
            AggregationBuilder aggregationBuilder = AggregationBuilders.max("maxAge").field("age");
            searchQueryBuilder.addAggregation(AggregationBuilders.terms("dogBreed").field("dogBreed.keyword").subAggregation(aggregationBuilder));
        }
        try {
            NativeSearchQuery searchQuery = searchQueryBuilder.build();
            log.info("searchGoods DSL:{}", searchQuery.getQuery());
            SearchHits<EsDogIndex> search = restTemplate.search(searchQuery, EsDogIndex.class);
            Aggregations aggregations = (Aggregations) search.getAggregations();
            Map<String, Aggregation> aggregationMap = aggregations.getAsMap();
            List list = this.convertToEsDogRelatedInfo(aggregationMap, dogSearchDTO);
            return list;
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 转换搜索结果为聚合商品展示信息
     *
     * @param aggregationMap 搜索结果
     * @return 聚合商品展示信息
     */
    private List convertToEsDogRelatedInfo(Map<String, Aggregation> aggregationMap,EsDogSearchDTO dogSearchDTO) {
        if (StringUtil.isNotBlank(dogSearchDTO.getGroupByAge())) {
            List<Map> ageList = new ArrayList<>();
            ParsedLongTerms ageCountTerms = (ParsedLongTerms) aggregationMap.get("ageCount");
            List<? extends Terms.Bucket> ageCountBuckets = ageCountTerms.getBuckets();
            if (ageCountBuckets != null && !ageCountBuckets.isEmpty()) {
                for (Terms.Bucket ageCountBucket : ageCountBuckets) {
                    //System.out.println("age字段是" + ageCountBucket.getKey() + ", ageCount是: " + ageCountBucket.getDocCount());
                    Map map = new HashMap();
                    map.put("age",ageCountBucket.getKey());
                    map.put("ageCount",ageCountBucket.getDocCount());
                    ageList.add(map);
                }
            }
            return ageList;
        }
        if (StringUtil.isNotBlank(dogSearchDTO.getGroupByAgeAndBreed())) {
            List<Map> mapList = new ArrayList<>();
            ParsedLongTerms ageCountTerms = (ParsedLongTerms) aggregationMap.get("ageCount");
            List<? extends Terms.Bucket> ageCountBuckets = ageCountTerms.getBuckets();
            if (ageCountBuckets != null && !ageCountBuckets.isEmpty()) {
                for (Terms.Bucket ageCountBucket : ageCountBuckets) {
                    String age = ageCountBucket.getKey().toString();
                    ParsedStringTerms breedCountTerms = ageCountBucket.getAggregations().get("breedCount");
                    List<? extends Terms.Bucket> breedCountBuckets = breedCountTerms.getBuckets();
                    for (Terms.Bucket breedCountBucket : breedCountBuckets) {
                        String dogBreed = breedCountBucket.getKey().toString();
                        long breedCount = breedCountBucket.getDocCount();
                        Map map = new HashMap();
                        map.put("age",age);
                        map.put("dogBreed",dogBreed);
                        map.put("breedCount",breedCount);
                        mapList.add(map);
                    }
                }
            }
            return mapList;
        }else{
            List<Map> mapList = new ArrayList<>();
            ParsedStringTerms ageCountTerms = (ParsedStringTerms) aggregationMap.get("dogBreed");
            List<? extends Terms.Bucket> ageCountBuckets = ageCountTerms.getBuckets();
            if (ageCountBuckets != null && !ageCountBuckets.isEmpty()) {
                for (Terms.Bucket ageCountBucket : ageCountBuckets) {
                    String age = ageCountBucket.getKey().toString();
                    ParsedMax parsedMax = ageCountBucket.getAggregations().get("maxAge");
                    Map map = new HashMap();
                    map.put("age",age);
                    map.put("dogBreed",parsedMax.getValue());
                    mapList.add(map);
                }
            }
            return mapList;
        }
    }

}
